package com.spark.ml

import org.apache.spark.mllib.clustering.LDA
import org.apache.spark.mllib.feature.{HashingTF, IDF}
import org.apache.spark.mllib.linalg.{SparseVector, Vector}
import org.apache.spark.rdd.RDD

/**
  * 中文分词,TF-IDF-LDA主题提取
  */
object ZH_AN extends MLBase {


    /**
      * 入口方法
      *
      * @param args
      */
    def main(args: Array[String]) {
        //如果没有参数传递则直接使用默认值
        setDefaultConf()
        if(args.length > 0){
            settingsConf(args(0))
        }
        //获取sparkContext实例
        val spark = getSparkInc()
        //加载mongodb指定的数据
        val mongoData = getMongoLoad(spark)
        //缓存加载出来的mongobb数据
        mongoData.cache()
        //转换rdd的格式,将数据分开
        val waitRdd = mongoData.map(data => {
            (data.getString("html"), data.getString("title"))
        })
        //调用函数,加载个人字典,再这里调用的目的是使字典只被加载一次
        //loadDict()
        //通过对document的Rdd操作,取出每个文本,先进行分词操作,然后将得到的词组list和并,用空格链接各个元素后交给tokenize方法去实现过滤和特征转换
        val documents = waitRdd.map(_._1).map(jieba).map(tokenize)
        //        val (wordsDF,sparkSql) = this.makeStopWordsRemover(spark,documents)
        //获取title的值
        val titles = waitRdd.map(_._2)
        //实例化hashingTF对象,这里调用的是spark.mllib库的hashingTF
        val hashingTF = new HashingTF()
        //通过对documents的value值使用flatMap的变换,统一value值到一起,得出了索引和word的一一对应的map关系,这里主要是利用了HashingTF的 indexOf方法
        //        wordsDF.as[Array]
        //        import sparkSql.implicits._
        //        val wordsRdd = wordsDF.as[String].map(x => x).rdd
        //        wordsDF.collect().foreach(println(_))
        val mapWords = documents.flatMap(x => x)
            .map(w => (hashingTF.indexOf(w), w))
            .collect
            .toMap
        //使用documents来进行hashingTF转换
        val tf = hashingTF.transform(documents)
        //将生成的单词语index对应关系的Map传播到各个worker上去,使得改值能在各个节点上被使用
        val bcWords = tf.context.broadcast(mapWords)
        tf.cache()
        //实例化IDF对象,调用fit方法训练IDF模型,并指出mindocs的值为2,表示至少在2个文当中出现过的词才会被计算
        val idf = new IDF(1).fit(tf)
        //使用hashingTF的内容来得到idf转换
        val tf_idf: RDD[Vector] = idf.transform(tf)
        //遍历生成的idf结果,得到每篇文章的前20个模糊主题的内容
        val r = tf_idf.map {
            case SparseVector(size, indices, values) =>
                val words = indices.map(index => bcWords.value.getOrElse(index, "null"))
                words.zip(values).sortBy(-_._2).take(20).toSeq
        }
        //将title和生成的结果组合,将结果写入到了本地文件系统中
        titles.zip(r).saveAsTextFile("/Users/xiaojun/Desktop/scalaLog/mongo_zhongwen_" + System.currentTimeMillis)
        //根据LDA的输入的需要,调用rdd的zipWithIndex方法将rdd标记后换位转换

        val rRdd = tf_idf.zipWithIndex().map(data => {
            (data._2, data._1)
        })
        //使用LDA方法实现对idf输出的结果进行聚类
        val ldaModel = new LDA().setK(3). //设置主题个数
            setDocConcentration(10). //设置文本
            setTopicConcentration(10). //设置主题
            setMaxIterations(10). //设置最大迭代次数,一般情况下迭代次数越多聚类越准确
            setSeed(0L). //设置随机种子
            setCheckpointInterval(10). //设置检查点
            setOptimizer("em"). //设置检验模式
            run(rRdd)

        //获取LDA聚类后的结果
        val desTopics = ldaModel.describeTopics(10)
        //循环打印结果
        desTopics.foreach(data => {
            print("Topic  :\n")
            for (i <- Range(0, 10)) {
                val tempIndex = data._1.toList(i)
                val tempWeight = data._2.toList(i)
                println(bcWords.value.getOrElse(tempIndex, "null") + "   " + tempWeight)
            }

        })
        //退出sparkContext,结束操作
        spark.stop()

    }

}
